#pragma once
#include <algorithm>
#include <memory>
#include <string>
#include <librdkafka/rdkafkacpp.h>
#include <thread>
#include <atomic>
#include <iostream>
#include <functional>


// #include "user_action.pb.h"


namespace lyt{
    class KafkaConsumer{
    
    public:
        using ptr = std::unique_ptr<KafkaConsumer>;
        using MessageCallback = std::function<void(std::unique_ptr<RdKafka::Message>)>;
        
        KafkaConsumer(const std::string& brokers,const std::string& group_id,const std::string& topic)
                    :_topic(topic),_running(false){
            std::string errstr;
            // 创建kafka全局配置对象
            RdKafka::Conf* conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
            // 1. 设置brokers地址，消费者必须知道从哪里拉取消息（即 broker 地址）
            conf->set("bootstrap.servers",brokers,errstr);
            // 2. 设置消费者组id，即相同组的消费者会协同消费topic
            conf->set("group.id",group_id,errstr);
            // 3. 设置 offset 策略为 earliest
            // 如果首次订阅，没有保存 offset，从最早的消息开始拉取
            conf->set("auto.offset.reset","earliest",errstr);
            // 4. 创建kafkaconsumer实例（绑定配置）
            _consumer.reset(RdKafka::KafkaConsumer::create(conf, errstr));
            if(!_consumer){
                std::cerr << "Failed to create KafkaConsumer: " << errstr << std::endl;
                throw std::runtime_error("KafkaConsumer init failed");
            }
            delete conf;
            // 订阅指定的topic（必须执行一次subscribe才能消费）
            if(_consumer->subscribe({_topic}) != RdKafka::ERR_NO_ERROR){
                throw std::runtime_error("KafkaConsumer init failed: " + errstr);
            }
        }
        ~KafkaConsumer(){
            stop();
            if(_consumer) _consumer->close();   // 当前消费者推出分区消费
        }

        void start(){
            _running = true;
            // 启动独立线程，不阻塞主服务流程
            _consume_thread = std::thread(&KafkaConsumer::consumeLoop,this);  
        }
        void stop(){
            _running = false;
            // 检查消费线程是否可加入状态（即已经启动并在运行）
            if(_consume_thread.joinable()){
                _consume_thread.join();
            }
        }
        KafkaConsumer& setMessageCallback(MessageCallback cb){
            _callback = std::move(cb);
            return *this;
        }
    private:
        void consumeLoop(){
            while(_running){
                // 间隔1s从kafka中拉取一次消息
                RdKafka::Message* msg = _consumer->consume(1000);
                if(!msg) continue;

                if(msg->err() == RdKafka::ERR_NO_ERROR){
                    if(_callback){
                        // 回调函数存在，那就将msg传递给回调函数，让回调去处理 
                        _callback(std::unique_ptr<RdKafka::Message>(msg));
                    }else{
                        // 不存在释放msg
                        delete msg;
                    }
                }else{
                    // 发生错误，非超时打印日志(因为超时情况是等待1s导致的返回信息，不能算错误)
                    if(msg->err() != RdKafka::ERR__TIMED_OUT){
                        std::cerr << "Kafka error:" << msg->errstr() << std::endl;
                    }
                    delete msg;
                }
            }
        }
    private:
        std::unique_ptr<RdKafka::KafkaConsumer> _consumer;  // Kafka消费者实例
        std::string _topic;                                 // 要订阅的topic名称
        std::thread _consume_thread;                        // 消费线程
        std::atomic<bool> _running;                         // 消费循环运行标志，线程安全
        MessageCallback _callback;                          // 外部传入的消息处理函数
    };
}